/**
 * MyListener.java
 *
 * @author: ZhuJiahui
 * @date: 2019/1/9 22:15
 */
package com.zhujiahui.sbkafka.second;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;

import java.util.Optional;

/**
 * @description:
 * @author: ZhuJiahui
 * @version: 1.0
 */
@Configuration
public class MyListener {

    private static final Logger LOGGER = LoggerFactory.getLogger(MyListener.class);

    public MyListener() {
    }

    @KafkaListener(topicPattern = "sbkafka.*")
    public void listen(ConsumerRecord<?, ?> record, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) {
        Optional<?> kafkaMessage = Optional.ofNullable(record.value());
        if (kafkaMessage.isPresent()) {
            Object message = kafkaMessage.get();
            LOGGER.info("------------------ message =topic：" + topic+ ", " + message);
        }
    }
}
